#ifndef __M_MESSAGE_H__
#define __M_MESSAGE_H__

#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h" 
#include <iostream>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <string>
#include <list>

namespace wsmq
{
    #define DATAFILE_SUFFIX ".mqd"
    #define TMPFILE_SUFFIX ".mqd.tmp"
    using MessagePtr = std::shared_ptr<wsmq::Message>;
    class MessageMapper
    {
    public:
        MessageMapper(std::string& basedir, const std::string& qname): _qname(qname)
        {
            if(basedir.back() != '/') basedir.push_back('/');
            _datafile = basedir + _qname + DATAFILE_SUFFIX;
            _tmpfile = basedir + _qname + TMPFILE_SUFFIX;
            if(FileHelper(basedir).exists() == false)
            {
                assert(FileHelper::createDirectory(basedir));
            }
            createMsgFile();
        }
        bool createMsgFile()
        {
            if(FileHelper(_datafile).exists() == true) // 导致[ERR][15:54:10][../mqserver/../mqcommon/mq_helper.hpp:120]      ./data/message/queue1.mqd 文件读取数据失败！！
            {
                return true;
            }
            bool ret = FileHelper::createFile(_datafile);
            if(ret == false)
            {
                DLOG("创建队列文件 %s 失败", _datafile.c_str());
                return false;
            }
            return true;
        }
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }
        bool insert(MessagePtr& msg)
        {
            return insert(_datafile, msg);
        }
        bool remove(MessagePtr& msg)
        {
            // 将msg的有效标志设置为'0'
            msg->mutable_payload()->set_valid("0");
            // 将msg序列化为字符串
            std::string body = msg->payload().SerializeAsString();
            if(body.size() != msg->length())
            {
                DLOG("不能修改文件中数据，因为新生成的数据和原数据长度不一致");
                return false;
            }
            //  将序列化后消息写入数据在文件中位置（覆盖原有数据）
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if(ret == false)
            {
                DLOG("向队列文件写入数据失败");
                return false;
            }
            return true;
        }
        std::list<MessagePtr> gc()
        {
            bool ret;
            std::list<MessagePtr> result;
            ret = load(result);
            if(ret == false)
            {
                DLOG("加载有效数据失败");
                return result;
            }
            DLOG("垃圾回收，得到有效消息数量：%ld", result.size());
            // 有效数据存储到临时文件
            FileHelper::createFile(_tmpfile);
            for(auto& msg : result)
            {
                DLOG("向临时文件写入有效数据 %s", msg->payload().body().c_str());
                ret = insert(_tmpfile, msg);
                if(ret == false)
                {
                    DLOG("向临时文件写入有效数据失败");
                    return result;
                }
            }
            DLOG("垃圾回收后，临时文件大小：%ld", FileHelper(_tmpfile).size());
            // 删除原文件
            ret = FileHelper::removeFile(_datafile);
            if(ret == false)
            {
                DLOG("删除原文件失败");
                return result;
            }
            // 修改临时文件名为原文件名
            ret = FileHelper(_tmpfile).rename(_datafile);
            if(ret == false)
            {
                DLOG("修改临时文件名为原文件名失败");
                return result;
            }
            // 返回全部有效数据
            return result;
        }
    private:
        bool insert(const std::string& filename, MessagePtr& msg)
        {
            // 新增数据添加到文件末尾
            // 进行消息序列化，得到格式化字符串消息
            std::string body = msg->payload().SerializeAsString();
            // 获取文件长度
            FileHelper helper(filename);
            size_t fsize = helper.size(); // 文件偏移量
            size_t msg_size = body.size(); // 消息长度
            bool ret = helper.write((char*)&msg_size, fsize, sizeof(size_t));
            if(ret == false)
            {
                DLOG("向队列文件写入消息长度失败");
                return false;
            }
            // 将数据写入文件对应位置
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
            if(ret == false)
            {
                DLOG("向队列文件写入消息数据失败");
                return false;
            }

            // 更新msg中实际存储信息
            msg->set_offset(fsize + sizeof(size_t));
            msg->set_length(body.size());
            return true;
        }
        bool load(std::list<MessagePtr>& result)
        {
            // 加载出文件中有效数据：按照数据格式：4字节消息长度|数据 读取数据，根据valid字段判断数据是否有效
            FileHelper helper(_datafile);
            size_t msg_size, offset = 0;
            size_t fsize = helper.size();
            DLOG("准备加载持久化数据，当前文件大小为：%ld", helper.size());
            bool ret;
            while(offset < fsize)
            {
                // ret = helper.read((char*)&msg_size, fsize, sizeof(size_t));--------------错了！！！！！！！
                ret = helper.read((char*)&msg_size, offset, sizeof(size_t));
                if(ret == false)
                {
                    DLOG("获取消息长度失败");
                    return false;
                }
                offset += sizeof(size_t);
                std::string msg_body(msg_size, '\0');
                ret = helper.read(&msg_body[0], offset, msg_size);
                if(ret == false)
                {
                    DLOG("读取消息数据失败");
                    return false;
                }
                offset += msg_size;
                MessagePtr pmsg = std::make_shared<Message>();
                pmsg->mutable_payload()->ParseFromString(msg_body);
                if(pmsg->payload().valid() == "0")
                {
                    DLOG("加载到无效消息: %s", pmsg->payload().body().c_str());
                    continue;
                }
                result.push_back(pmsg);
            }
            return true;
        }
        
    private:
        std::string _qname;
        std::string _datafile;
        std::string _tmpfile;

    };
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string& basedir, const std::string& qname): _mapper(basedir, qname),
            _qname(qname), _valid_count(0), _total_count(0)
        {}
        bool recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc(); // 恢复历史消息数据
            for(auto& msg : _msgs)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            _total_count = _valid_count = _msgs.size();
            return true;
        }
        bool insert(const BasicProperties* bp, const std::string& body, bool queue_is_durable)
        {
            // 构造消息对象
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            if(bp != nullptr) // 传入的消息属性不为空
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断消息是否需要持久化
            if(msg->payload().properties().delivery_mode() == wsmq::DeliveryMode::DURABLE)
            {
                // 进行持久化
                msg->mutable_payload()->set_valid("1");
                bool ret = _mapper.insert(msg);
                if(ret == false)
                {
                    DLOG("持久化消息失败");
                    return false;
                }
                _valid_count += 1;
                _total_count += 1;
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            // 在内存中的管理
            _msgs.push_back(msg);
            return true;
        }
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_msgs.size() == 0) {
                return MessagePtr();
            }
            // 获取队首消息：从_msgs中取出数据
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // 将消息添加到待确认消息中，一旦确认，就可以删除
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }

        bool remove(const std::string& msg_id) // 每次删除消息后，根据情况判断是否需要垃圾回收
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 从待确认消息中查找到要删除的消息
            auto it = _waitack_msgs.find(msg_id);
            if(it == _waitack_msgs.end())
            {
                return true;
            }
            // 根据消息的持久化模式判断是否进行过持久化
            if(it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 删除持久化消息
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count -= 1;
                gc();
            }
            // 删除内存中数据
            _waitack_msgs.erase(msg_id);
            return true;
        }
        size_t getable_count() // 待推送消息数量
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }
        size_t total_count() // 持久化消息历史总数
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        size_t durable_count() // 当前有效持久化消息数量
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }
        size_t waitack_count() // 待确认消息数量
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _valid_count = _total_count = 0;
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
        }
    private:
        bool GCCheck()
        {
            // 消息总数且有效消息比例小于50%则进行垃圾回收
            if(_total_count > 2000 && _valid_count * 1.0 / _total_count <= 0.5)
            {
                return true;
            }
            return false;
        }
        void gc()
        {
            // 进行垃圾回收，获取垃圾回收后的有效消息链表
            if(GCCheck() == false) return ;
            std::list<MessagePtr> msgs = _mapper.gc();
            for(auto& msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if(it == _durable_msgs.end())
                {
                    DLOG("垃圾回收后，发现未管理的有效持久化消息");
                    _msgs.push_back(it->second);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), it->second));
                    continue;
                }
                // 更新每条消息的实际存储位置
                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            // 更新当前有效数据条数 && 总消息条数
            _total_count = _valid_count = msgs.size();
        }
    private:
        std::mutex _mutex;
        std::string _qname;
        size_t _valid_count;
        size_t _total_count;
        MessageMapper _mapper;
        std::list<MessagePtr> _msgs; // 待推送消息
        std::unordered_map<std::string, MessagePtr> _durable_msgs;
        std::unordered_map<std::string, MessagePtr> _waitack_msgs;
    };
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string& basedir): _basedir(basedir)
        {}
        void clear()
        {
            for(auto msg : _queue_msgs)
            {
                msg.second->clear();
            }
        }
        void initQueueMessage(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end()) 
                {
                    return ;
                }
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            qmp->recovery(); // 此操作本身是线程安全的（已经加锁保护了），不需要放在临界区
        }
        void destroyQueueMessage(const std::string& qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) 
                {
                    return ;
                }
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            qmp->clear(); // 本身就是被加锁保护的，不需要外部再加锁了，否则降低效率
        }
        MessagePtr front(const std::string& qname)
        {
            QueueMessage::ptr qmp = getQueueMessageHandle(qname);
            if(qmp == nullptr)
            {
                DLOG("获取队首消息失败");
                return nullptr;
            }
            return qmp->front();
        }
        void ack(const std::string& qname, const std::string& msg_id)
        {
            QueueMessage::ptr qmp = getQueueMessageHandle(qname);
            if(qmp == nullptr)
            {
                DLOG("确认队列消息 %s 失败", msg_id.c_str());
                return ;
            }
            qmp->remove(msg_id);
            return ;
        }
        bool insert(const std::string& qname, const BasicProperties* bp,
             const std::string& body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp = getQueueMessageHandle(qname);
            if(qmp == nullptr)
            {
                DLOG("向队列 %s 插入消息失败", qname.c_str());
                return false;
            }
            return qmp->insert(bp, body, queue_is_durable);
        }

        size_t getable_count(const std::string& qname)
        {
            QueueMessage::ptr qmp = getQueueMessageHandle(qname);
            if(qmp == nullptr)
            {
                DLOG("获取队列 %s 待推送消息数量失败", qname.c_str());
                return 0;
            }
            return qmp->getable_count();
        }
        size_t total_count(const std::string& qname)
        {
            QueueMessage::ptr qmp = getQueueMessageHandle(qname);
            if(qmp == nullptr)
            {
                DLOG("获取队列 %s 总体消息数量失败", qname.c_str());
                return 0;
            }
            return qmp->total_count();
        }
        size_t durable_count(const std::string& qname)
        {
            QueueMessage::ptr qmp = getQueueMessageHandle(qname);
            if(qmp == nullptr)
            {
                DLOG("获取队列 %s 持久化消息数量失败", qname.c_str());
                return 0;
            }
            return qmp->durable_count();
        }
        size_t waitack_count(const std::string& qname)
        {
            QueueMessage::ptr qmp = getQueueMessageHandle(qname);
            if(qmp == nullptr)
            {
                DLOG("获取队列 %s 待确认消息数量失败", qname.c_str());
                return 0;
            }
            return qmp->waitack_count();
        }
    private:
        // 由于需要频繁查找队列名对应的操作句柄，所以封装成函数
        QueueMessage::ptr getQueueMessageHandle(const std::string& qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if(it == _queue_msgs.end())
            {
                DLOG("未找到队列 %s 操作句柄", qname.c_str());
                return QueueMessage::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
    };
}

#endif